/****************************************************************
 * Licensed to the Apache Software Foundation (ASF) under one   *
 * or more contributor license agreements.  See the NOTICE file *
 * distributed with this work for additional information        *
 * regarding copyright ownership.  The ASF licenses this file   *
 * to you under the Apache License, Version 2.0 (the            *
 * "License"); you may not use this file except in compliance   *
 * with the License.  You may obtain a copy of the License at   *
 *                                                              *
 *   http://www.apache.org/licenses/LICENSE-2.0                 *
 *                                                              *
 * Unless required by applicable law or agreed to in writing,   *
 * software distributed under the License is distributed on an  *
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY       *
 * KIND, either express or implied.  See the License for the    *
 * specific language governing permissions and limitations      *
 * under the License.                                           *
 ****************************************************************/
package org.apache.james.mailbox.cassandra.mail;
import java.io.BufferedInputStream;
import java.util.logging.Level;
import java.util.logging.Logger;
import org.apache.cassandra.thrift.InvalidRequestException;
import org.apache.cassandra.thrift.KeySlice;
import org.apache.cassandra.thrift.KeyRange;
import org.apache.cassandra.thrift.NotFoundException;
import org.apache.cassandra.thrift.SliceRange;
import org.apache.cassandra.thrift.SlicePredicate;
import org.apache.cassandra.thrift.ConsistencyLevel;
import org.apache.cassandra.thrift.TimedOutException;
import org.apache.cassandra.thrift.UnavailableException;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.charset.CharacterCodingException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import javax.mail.Flags;
import org.apache.cassandra.thrift.Cassandra;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.thrift.ColumnParent;
import org.apache.cassandra.thrift.ColumnPath;
import org.apache.cassandra.thrift.ColumnOrSuperColumn;
import org.apache.cassandra.thrift.TBinaryProtocol;
import org.apache.thrift.TException;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.transport.TFramedTransport;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransport;
import org.apache.thrift.transport.TTransportException;
import org.apache.james.mailbox.MailboxException;
import org.apache.james.mailbox.MailboxSession;
import org.apache.james.mailbox.MessageMetaData;
import org.apache.james.mailbox.MessageRange;
import org.apache.james.mailbox.MessageRange.Type;
import org.apache.james.mailbox.UpdatedFlags;
import org.apache.james.mailbox.cassandra.CNames;
import org.apache.james.mailbox.cassandra.CUtils;
import org.apache.james.mailbox.cassandra.FlagConvertor;
import org.apache.james.mailbox.cassandra.io.ChunkOutputStream;
import org.apache.james.mailbox.store.SimpleMessageMetaData;
import org.apache.james.mailbox.store.transaction.NonTransactionalMapper;
import org.apache.james.mailbox.store.mail.MessageMapper;
import org.apache.james.mailbox.store.mail.MessageMapper.FetchType;
import org.apache.james.mailbox.store.mail.MessageMapper.MessageCallback;
import org.apache.james.mailbox.store.mail.ModSeqProvider;
import org.apache.james.mailbox.store.mail.UidProvider;
import org.apache.james.mailbox.store.mail.model.Mailbox;
import org.apache.james.mailbox.store.mail.model.Message;




           
        


/**
 * 
 * HBase implementation of a {@link MessageMapper}. 
 * I don't know if this class is thread-safe! Asume it is not!
 * 
 */
public class CMessageMapper extends NonTransactionalMapper implements MessageMapper<UUID> {

    
    private final MailboxSession mailboxSession;
    private final UidProvider<UUID> uidProvider;
    private final ModSeqProvider<UUID> modSeqProvider;
    private Map<Long, MessageMetaData> data;
    private long count;
    Cassandra.Client client;
    public CMessageMapper(final MailboxSession session,
            final UidProvider<UUID> uidProvider,
            ModSeqProvider<UUID> modSeqProvider) {
        this.mailboxSession = session;
        this.modSeqProvider = modSeqProvider;
        this.uidProvider = uidProvider;
        
    }
               	
    @Override
    public void endRequest() { }

    @Override
    public void findInMailbox(Mailbox<UUID> mailbox, MessageRange set, FetchType fType, MessageCallback<UUID> callback) throws MailboxException {
        try {
            
         TTransport transport = new TFramedTransport(new TSocket("localhost", 9160));
         TProtocol protocol = new TBinaryProtocol(transport);
         transport.open();
         client = new Cassandra.Client(protocol);
         client.set_keyspace(CNames.keyspace); 
            
            
            
            
            List<Message<UUID>> results;
            long from = set.getUidFrom();
            final long to = set.getUidTo();
            final int batchSize = set.getBatchSize();
            final Type type = set.getType();

            // when batch is specified fetch data in chunks and send back in
            // batches
            do {
                switch (type) {
                    default:
                    case ALL:
                        results = findMessagesInMailbox(mailbox, batchSize, false);
                        break;
                    case FROM:
                        results = findMessagesInMailboxAfterUID(mailbox, from, batchSize, false);
                        break;
                    case ONE:
                        results = findMessagesInMailboxWithUID(mailbox, from, false);
                        break;
                    case RANGE:
                        results = findMessagesInMailboxBetweenUIDs(mailbox, from, to, batchSize, false);
                        break;
                }

                if (results.size() > 0) {
                    callback.onMessages(results);

                    // move the start UID behind the last fetched message UID
                    from = results.get(results.size() - 1).getUid() + 1;
                }

            } while (results.size() > 0 && batchSize > 0);

        } catch (NotFoundException ex) {
            Logger.getLogger(CMessageMapper.class.getName()).log(Level.SEVERE, null, ex);
        } catch (UnavailableException ex) {
            Logger.getLogger(CMessageMapper.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TimedOutException ex) {
            Logger.getLogger(CMessageMapper.class.getName()).log(Level.SEVERE, null, ex);
        } catch (InvalidRequestException ex) {
            Logger.getLogger(CMessageMapper.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TException ex) {
            Logger.getLogger(CMessageMapper.class.getName()).log(Level.SEVERE, null, ex);
        
        } catch (IOException e) {
            throw new MailboxException("Search of MessageRange " + set + " failed in mailbox " + mailbox, e);
        }
    }

    private List<Message<UUID>> findMessagesInMailbox(Mailbox<UUID> mailbox, int batchSize, boolean flaggedForDelete) throws IOException, InvalidRequestException, NotFoundException, UnavailableException, TimedOutException, TException {
        
        TTransport transport = new TFramedTransport(new TSocket("localhost", 9160));
        TProtocol protocol = new TBinaryProtocol(transport);
        transport.open();
        client = new Cassandra.Client(protocol);
        client.set_keyspace(CNames.keyspace); 
        
        
           List<Message<UUID>> messageList = new ArrayList<Message<UUID>>();
           ColumnPath c=new ColumnPath();
           ColumnOrSuperColumn d = new ColumnOrSuperColumn();
           byte[] b=null;
      
           c.setColumn_family(ByteBufferUtil.string(CNames.MESSAGES_META_CF));
           c.setColumn(FlagConvertor.FLAGS_DELETED); 
           
           ColumnParent messages= new ColumnParent(ByteBufferUtil.string(CNames.MESSAGES_META_CF)); 
           SlicePredicate predicate = new SlicePredicate();
           predicate.setSlice_range(new SliceRange(ByteBuffer.wrap(new byte[0]), ByteBuffer.wrap(new byte[0]), false, 100));
           KeyRange keyRange = new KeyRange(100);
           
           keyRange.setStart_key(CUtils.messageRowKey(mailbox.getMailboxId(),Long.MAX_VALUE));
            keyRange.setEnd_key(CUtils.messageRowKey(mailbox.getMailboxId(), 0));
            List<KeySlice> keySlices = client.get_range_slices(messages, predicate, keyRange, ConsistencyLevel.ONE);
       
            count = batchSize > 0 ? batchSize : Long.MAX_VALUE;
                
            for (KeySlice ks : keySlices) {
              if (!flaggedForDelete) {
                if(count > 0) {
                    messageList.add(CUtils.messageMetaFromResult((ks.getKey()).toString(), CNames.keyspace));
                    count--;
                }
              }else
                {
                  d = client.get(ByteBufferUtil.bytes(ks.getKey().toString()), c, ConsistencyLevel.ONE);
                  b = d.getColumn().getValue();
                 if(ByteBufferUtil.compare(ByteBufferUtil.bytes(b.toString()), ByteBufferUtil.getArray(CNames.MARKER_MISSING))==0){
                        if(count > 0) {
                        messageList.add(CUtils.messageMetaFromResult((ks.getKey()).toString(), CNames.keyspace));
                        count--;
                        }
                 }
        }
        // we store uids in reverse order, we send them ascending
        Collections.reverse(messageList);
        return messageList;
        }
            return messageList;
            
    }

    private List<Message<UUID>> findMessagesInMailboxWithUID(Mailbox<UUID> mailbox, final long messageUid, final boolean flaggedForDelete) throws IOException, InvalidRequestException, NotFoundException, UnavailableException, TimedOutException, TException {
        
        
        TTransport transport = new TFramedTransport(new TSocket("localhost", 9160));
        TProtocol protocol = new TBinaryProtocol(transport);
        transport.open();
        client = new Cassandra.Client(protocol);
        client.set_keyspace(CNames.keyspace); 
        
        
        
        
        List<Message<UUID>> messageList = new ArrayList<Message<UUID>>();
           ColumnPath c=new ColumnPath();
           ColumnOrSuperColumn d = new ColumnOrSuperColumn();
           c.setColumn_family(ByteBufferUtil.string(CNames.MESSAGES_META_CF));
           c.setColumn(FlagConvertor.FLAGS_DELETED); 
           byte[] b = CUtils.messageRowKey(mailbox.getMailboxId(), messageUid);
           
       Message<UUID> message = null;
        if (!flaggedForDelete) {
            message = CUtils.messageMetaFromResult(b.toString(), CNames.keyspace);
            messageList.add(message);
       }else
        {
            d = client.get(ByteBufferUtil.bytes(b.toString()), c, ConsistencyLevel.ONE);
            b = d.getColumn().getValue();
            if(ByteBufferUtil.compare(ByteBufferUtil.bytes(b.toString()),ByteBufferUtil.getArray(CNames.MARKER_MISSING))==0){
                 message = CUtils.messageMetaFromResult(b.toString(), CNames.keyspace);
                 messageList.add(message);   
                }
        }
       
        return messageList;
    }

    private List<Message<UUID>> findMessagesInMailboxAfterUID(Mailbox<UUID> mailbox, final long from, final int batchSize, final boolean flaggedForDelete) throws IOException {        
        List<Message<UUID>> messageList = new ArrayList<Message<UUID>>();
       
        return messageList;
    }

    private List<Message<UUID>> findMessagesInMailboxBetweenUIDs(Mailbox<UUID> mailbox, final long from, final long to, final int batchSize, final boolean flaggedForDelete) throws IOException {
        List<Message<UUID>> messageList = new ArrayList<Message<UUID>>();
       
        return messageList;
    }

    @Override
    public Map<Long, MessageMetaData> expungeMarkedForDeletionInMailbox(Mailbox<UUID> mailbox, MessageRange set) throws MailboxException {
        try {
           // Map<Long, MessageMetaData> data;
            List<Message<UUID>> results = null;
            final long from = set.getUidFrom();
            final long to = set.getUidTo();

            switch (set.getType()) {
                case ONE:
            try {
                results = findMessagesInMailboxWithUID(mailbox, from, true);
            } catch (InvalidRequestException ex) {
                Logger.getLogger(CMessageMapper.class.getName()).log(Level.SEVERE, null, ex);
            } catch (NotFoundException ex) {
                Logger.getLogger(CMessageMapper.class.getName()).log(Level.SEVERE, null, ex);
            } catch (UnavailableException ex) {
                Logger.getLogger(CMessageMapper.class.getName()).log(Level.SEVERE, null, ex);
            } catch (TimedOutException ex) {
                Logger.getLogger(CMessageMapper.class.getName()).log(Level.SEVERE, null, ex);
            } catch (TException ex) {
                Logger.getLogger(CMessageMapper.class.getName()).log(Level.SEVERE, null, ex);
            }
     
            data = createMetaData(results);
                    deleteDeletedMessagesInMailboxWithUID(mailbox, from);
                    break;
                case RANGE:
                    results = findMessagesInMailboxBetweenUIDs(mailbox, from, to, -1, true);
                    data = createMetaData(results);
                    deleteDeletedMessagesInMailboxBetweenUIDs(mailbox, from, to);
                    break;
                case FROM:
                    results = findMessagesInMailboxAfterUID(mailbox, from, -1, true);
                    data = createMetaData(results);
                    deleteDeletedMessagesInMailboxAfterUID(mailbox, from);
                    break;
                default:
                case ALL:
                    results = findMessagesInMailbox(mailbox, -1, true);
                    data = createMetaData(results);
                    deleteDeletedMessagesInMailbox(mailbox);
                    break;
            }

            
        } catch (InvalidRequestException ex) {
            Logger.getLogger(CMessageMapper.class.getName()).log(Level.SEVERE, null, ex);
        } catch (NotFoundException ex) {
            Logger.getLogger(CMessageMapper.class.getName()).log(Level.SEVERE, null, ex);
        } catch (UnavailableException ex) {
            Logger.getLogger(CMessageMapper.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TimedOutException ex) {
            Logger.getLogger(CMessageMapper.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TException ex) {
            Logger.getLogger(CMessageMapper.class.getName()).log(Level.SEVERE, null, ex);
        } catch (IOException e) {
            throw new MailboxException("Search of MessageRange " + set + " failed in mailbox " + mailbox, e);
        }
        
        return data;
    }

    @Override
    public long countMessagesInMailbox(Mailbox<UUID> mailbox) throws MailboxException {
        
        try{
        
        TTransport transport = new TFramedTransport(new TSocket("localhost", 9160));
        TProtocol protocol = new TBinaryProtocol(transport);
        transport.open();
        client = new Cassandra.Client(protocol);
        client.set_keyspace(CNames.keyspace); 
            
       
            count = 0;
            ColumnPath c =new ColumnPath();
            c.setColumn_family(ByteBufferUtil.string(CNames.MAILBOX_CF));
            c.setColumn(CNames.MAILBOX_MESSAGE_COUNT);
            ColumnOrSuperColumn s = client.get(ByteBufferUtil.bytes((CUtils.mailboxRowKey(mailbox.getMailboxId())).toString()), c, ConsistencyLevel.ONE);
            count=ByteBufferUtil.toLong(ByteBufferUtil.bytes(s.getColumn().getValue().toString()));
           
        } catch (NotFoundException ex) {
            Logger.getLogger(CMessageMapper.class.getName()).log(Level.SEVERE, null, ex);
        } catch (UnavailableException ex) {
            Logger.getLogger(CMessageMapper.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TimedOutException ex) {
            Logger.getLogger(CMessageMapper.class.getName()).log(Level.SEVERE, null, ex);
        } catch (InvalidRequestException ex) {
            Logger.getLogger(CMessageMapper.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TException ex) {
            Logger.getLogger(CMessageMapper.class.getName()).log(Level.SEVERE, null, ex);
        
        } catch (IOException e) {
            throw new MailboxException("Count of messages failed in mailbox " + mailbox, e);
        } 
        
         return count;
    }

    @Override
    public long countUnseenMessagesInMailbox(Mailbox<UUID> mailbox) throws MailboxException {
        /* TODO: see if it is possible to store the number of unseen messages in the mailbox table
         * and just return that value with a Get and kepp it up to date.
         */
      
        try {
            
        TTransport transport = new TFramedTransport(new TSocket("localhost", 9160));
        TProtocol protocol = new TBinaryProtocol(transport);
        transport.open();
        client = new Cassandra.Client(protocol);
        client.set_keyspace(CNames.keyspace); 
             
            count = 0;
            ColumnPath c=new ColumnPath();
           ColumnOrSuperColumn d = new ColumnOrSuperColumn();
           c.setColumn_family(ByteBufferUtil.string(CNames.MESSAGES_META_CF));
           c.setColumn(FlagConvertor.FLAGS_SEEN); 
           
           ColumnParent messages= new ColumnParent(ByteBufferUtil.string(CNames.MESSAGES_META_CF)); 
           SlicePredicate predicate = new SlicePredicate();
           predicate.setSlice_range(new SliceRange(ByteBuffer.wrap(new byte[0]), ByteBuffer.wrap(new byte[0]), false, 100));
           KeyRange keyRange = new KeyRange(100);
           keyRange.setStart_key(CUtils.messageRowKey(mailbox.getMailboxId(),Long.MAX_VALUE));
            keyRange.setEnd_key(CUtils.messageRowKey(mailbox.getMailboxId(), 0));
            List<KeySlice> keySlices = client.get_range_slices(messages, predicate, keyRange, ConsistencyLevel.ONE);
            
            for (KeySlice ks : keySlices) {
                d = client.get(ByteBufferUtil.bytes(ks.getKey().toString()), c, ConsistencyLevel.ONE);
                byte[] b = d.getColumn().getValue();
                if(ByteBufferUtil.compare(ByteBufferUtil.bytes(b.toString()), ByteBufferUtil.getArray(CNames.MARKER_MISSING))==0){
                   count++;
               }
            }
 
            
        } catch (NotFoundException ex) {
            Logger.getLogger(CMessageMapper.class.getName()).log(Level.SEVERE, null, ex);
        } catch (UnavailableException ex) {
            Logger.getLogger(CMessageMapper.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TimedOutException ex) {
            Logger.getLogger(CMessageMapper.class.getName()).log(Level.SEVERE, null, ex);
        } catch (InvalidRequestException ex) {
            Logger.getLogger(CMessageMapper.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TException ex) {
            Logger.getLogger(CMessageMapper.class.getName()).log(Level.SEVERE, null, ex);
        } catch (IOException e) {
            throw new MailboxException("Search of first unseen message failed in mailbox " + mailbox, e);
        }
        
        return count;
    }

    @Override
    public void delete(Mailbox<UUID> mailbox, Message<UUID> message) throws MailboxException {
        //TODO: maybe switch to checkAndDelete
        
        try {
            
        TTransport transport = new TFramedTransport(new TSocket("localhost", 9160));
        TProtocol protocol = new TBinaryProtocol(transport);
        transport.open();
        client = new Cassandra.Client(protocol);
        client.set_keyspace(CNames.keyspace); 
              
           ColumnPath c=new ColumnPath();
           ColumnOrSuperColumn d = new ColumnOrSuperColumn();
           
           c.setColumn_family(ByteBufferUtil.string(CNames.MESSAGES_META_CF));
           
           c.setColumn(CNames.MESSAGE_BODY_OCTETS);
           d= client.get(ByteBufferUtil.bytes(CUtils.messageRowKey(message).toString()), c, ConsistencyLevel.ONE);
           client.remove(ByteBufferUtil.bytes(CUtils.messageRowKey(message).toString()), c, d.getColumn().getTimestamp(), ConsistencyLevel.ONE);
         
          c.setColumn(CNames.MESSAGE_CONTENT_OCTETS);
           d= client.get(ByteBufferUtil.bytes(CUtils.messageRowKey(message).toString()), c, ConsistencyLevel.ONE);
           client.remove(ByteBufferUtil.bytes(CUtils.messageRowKey(message).toString()), c, d.getColumn().getTimestamp(), ConsistencyLevel.ONE);
           
           c.setColumn(CNames.MESSAGE_INTERNALDATE);
           d= client.get(ByteBufferUtil.bytes(CUtils.messageRowKey(message).toString()), c, ConsistencyLevel.ONE);
           client.remove(ByteBufferUtil.bytes(CUtils.messageRowKey(message).toString()), c, d.getColumn().getTimestamp(), ConsistencyLevel.ONE);
           
           c.setColumn(CNames.MESSAGE_MEDIA_TYPE);
           d= client.get(ByteBufferUtil.bytes(CUtils.messageRowKey(message).toString()), c, ConsistencyLevel.ONE);
           client.remove(ByteBufferUtil.bytes(CUtils.messageRowKey(message).toString()), c, d.getColumn().getTimestamp(), ConsistencyLevel.ONE);
           
           c.setColumn(CNames.MESSAGE_MODSEQ);
           d= client.get(ByteBufferUtil.bytes(CUtils.messageRowKey(message).toString()), c, ConsistencyLevel.ONE);
           client.remove(ByteBufferUtil.bytes(CUtils.messageRowKey(message).toString()), c, d.getColumn().getTimestamp(), ConsistencyLevel.ONE);

           c.setColumn(CNames.MESSAGE_SUB_TYPE);
           d= client.get(ByteBufferUtil.bytes(CUtils.messageRowKey(message).toString()), c, ConsistencyLevel.ONE);
           client.remove(ByteBufferUtil.bytes(CUtils.messageRowKey(message).toString()), c, d.getColumn().getTimestamp(), ConsistencyLevel.ONE);
           
           c.setColumn(CNames.MESSAGE_TEXT_LINE_COUNT);
           d= client.get(ByteBufferUtil.bytes(CUtils.messageRowKey(message).toString()), c, ConsistencyLevel.ONE);
           client.remove(ByteBufferUtil.bytes(CUtils.messageRowKey(message).toString()), c, d.getColumn().getTimestamp(), ConsistencyLevel.ONE);
           
           c.setColumn(FlagConvertor.FLAGS_ANSWERED);
           d= client.get(ByteBufferUtil.bytes(CUtils.messageRowKey(message).toString()), c, ConsistencyLevel.ONE);
           client.remove(ByteBufferUtil.bytes(CUtils.messageRowKey(message).toString()), c, d.getColumn().getTimestamp(), ConsistencyLevel.ONE);
           
           c.setColumn(FlagConvertor.FLAGS_DELETED);
           d= client.get(ByteBufferUtil.bytes(CUtils.messageRowKey(message).toString()), c, ConsistencyLevel.ONE);
           client.remove(ByteBufferUtil.bytes(CUtils.messageRowKey(message).toString()), c, d.getColumn().getTimestamp(), ConsistencyLevel.ONE);
           
           c.setColumn(FlagConvertor.FLAGS_DRAFT);
           d= client.get(ByteBufferUtil.bytes(CUtils.messageRowKey(message).toString()), c, ConsistencyLevel.ONE);
           client.remove(ByteBufferUtil.bytes(CUtils.messageRowKey(message).toString()), c, d.getColumn().getTimestamp(), ConsistencyLevel.ONE);
           
           c.setColumn(FlagConvertor.FLAGS_FLAGGED);
           d= client.get(ByteBufferUtil.bytes(CUtils.messageRowKey(message).toString()), c, ConsistencyLevel.ONE);
           client.remove(ByteBufferUtil.bytes(CUtils.messageRowKey(message).toString()), c, d.getColumn().getTimestamp(), ConsistencyLevel.ONE);
           
           c.setColumn(FlagConvertor.FLAGS_RECENT);
           d= client.get(ByteBufferUtil.bytes(CUtils.messageRowKey(message).toString()), c, ConsistencyLevel.ONE);
           client.remove(ByteBufferUtil.bytes(CUtils.messageRowKey(message).toString()), c, d.getColumn().getTimestamp(), ConsistencyLevel.ONE);
           
           c.setColumn(FlagConvertor.FLAGS_SEEN);
           d= client.get(ByteBufferUtil.bytes(CUtils.messageRowKey(message).toString()), c, ConsistencyLevel.ONE);
           client.remove(ByteBufferUtil.bytes(CUtils.messageRowKey(message).toString()), c, d.getColumn().getTimestamp(), ConsistencyLevel.ONE);
           
           c.setColumn(FlagConvertor.FLAGS_USER);
           d= client.get(ByteBufferUtil.bytes(CUtils.messageRowKey(message).toString()), c, ConsistencyLevel.ONE);
           client.remove(ByteBufferUtil.bytes(CUtils.messageRowKey(message).toString()), c, d.getColumn().getTimestamp(), ConsistencyLevel.ONE);
    
           
           c.setColumn_family(ByteBufferUtil.string(CNames.MESSAGE_DATA_BODY));

           c.setColumn(ByteBufferUtil.bytes(CNames.MESSAGES_DATA));
           d= client.get(ByteBufferUtil.bytes(CUtils.messageRowKey(message).toString()), c, ConsistencyLevel.ONE);
           client.remove(ByteBufferUtil.bytes(CUtils.messageRowKey(message).toString()), c, d.getColumn().getTimestamp(), ConsistencyLevel.ONE);
           
           c.setColumn_family(ByteBufferUtil.string(CNames.MESSAGE_DATA_HEADERS));
           
           c.setColumn(ByteBufferUtil.bytes(CNames.MESSAGES_DATA_HEADER));
           d= client.get(ByteBufferUtil.bytes(CUtils.messageRowKey(message).toString()), c, ConsistencyLevel.ONE);
           client.remove(ByteBufferUtil.bytes(CUtils.messageRowKey(message).toString()), c, d.getColumn().getTimestamp(), ConsistencyLevel.ONE);
           
           c.setColumn_family(ByteBufferUtil.string(CNames.MAILBOX_CF));
           c.setColumn(CNames.MAILBOX_MESSAGE_COUNT);
           d= client.get(ByteBuffer.wrap(CUtils.mailboxRowKey(mailbox.getMailboxId())), c, ConsistencyLevel.ONE);
           byte[] a = d.getColumn().getValue();
                      d.getColumn().setValue(ByteBufferUtil.bytes(ByteBufferUtil.toLong(ByteBufferUtil.bytes(a.toString())) - 1));
           
        } catch (NotFoundException ex) {
            Logger.getLogger(CMessageMapper.class.getName()).log(Level.SEVERE, null, ex);
        } catch (UnavailableException ex) {
            Logger.getLogger(CMessageMapper.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TimedOutException ex) {
            Logger.getLogger(CMessageMapper.class.getName()).log(Level.SEVERE, null, ex);
        } catch (InvalidRequestException ex) {
            Logger.getLogger(CMessageMapper.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TException ex) {
            Logger.getLogger(CMessageMapper.class.getName()).log(Level.SEVERE, null, ex);
        
        } catch (IOException ex) {
            throw new MailboxException("Delete of message " + message + " failed in mailbox " + mailbox, ex);
        } 

    }

    @Override
    public Long findFirstUnseenMessageUid(Mailbox<UUID> mailbox) throws MailboxException {
      return null;
    }

    @Override
    public List<Long> findRecentMessageUidsInMailbox(Mailbox<UUID> mailbox) throws MailboxException {
       return null;
    }

    /*
     * (non-Javadoc)
     * @see org.apache.james.mailbox.store.mail.MessageMapper#add(org.apache.james.mailbox.store.mail.model.Mailbox, org.apache.james.mailbox.store.mail.model.Message)
     */
    @Override
    public MessageMetaData add(Mailbox<UUID> mailbox, Message<UUID> message) throws MailboxException {
        message.setUid(uidProvider.nextUid(mailboxSession, mailbox));
        // if a mailbox does not support mod-sequences the provider may be null
        if (modSeqProvider != null) {
            message.setModSeq(modSeqProvider.nextModSeq(mailboxSession, mailbox));
        }
        MessageMetaData data1 = save(mailbox, message);

        return data1;
    }

    /*
     * (non-Javadoc)
     * @see org.apache.james.mailbox.store.mail.MessageMapper#updateFlags(org.apache.james.mailbox.store.mail.model.Mailbox, javax.mail.Flags, boolean, boolean, org.apache.james.mailbox.MessageRange)
     */
 @Override
    public Iterator<UpdatedFlags> updateFlags(final Mailbox<UUID> mailbox, final Flags flags, final boolean value, final boolean replace, MessageRange set) throws MailboxException {

        final List<UpdatedFlags> updatedFlags = new ArrayList<UpdatedFlags>();

        findInMailbox(mailbox, set, FetchType.Metadata, new MessageCallback<UUID>() {

            @Override
            public void onMessages(List<Message<UUID>> members) throws MailboxException {
                ColumnPath c=new ColumnPath();
                ColumnOrSuperColumn d = new ColumnOrSuperColumn();
                long modSeq = -1;
                if (members.isEmpty() == false) {
                    // if a mailbox does not support mod-sequences the provider may be null
                    if (modSeqProvider != null) {
                        modSeq = modSeqProvider.nextModSeq(mailboxSession, mailbox);
                    }
                }
                try {
                    c.setColumn_family(ByteBufferUtil.string(CNames.MESSAGES_META_CF));
                    c.setColumn(CNames.MESSAGE_MODSEQ);
                    Flags originalFlags, current;
                    
                    for (final Message<UUID> member : members) {
                        
                        originalFlags = member.createFlags();
                        current = null;
                        if (replace) {
                            current = new Flags(flags);
                        } else {
                            current = member.createFlags();
                            if (value) {
                                current.add(flags);
                            } else {
                                current.remove(flags);
                            }
                        }
                        try {
                            client = CUtils.flagsToPut(member, current);
                        } catch (TTransportException ex) {
                            Logger.getLogger(CMessageMapper.class.getName()).log(Level.SEVERE, null, ex);
                        } catch (InvalidRequestException ex) {
                            Logger.getLogger(CMessageMapper.class.getName()).log(Level.SEVERE, null, ex);
                        } catch (TException ex) {
                            Logger.getLogger(CMessageMapper.class.getName()).log(Level.SEVERE, null, ex);
                        } catch (CharacterCodingException ex) {
                            Logger.getLogger(CMessageMapper.class.getName()).log(Level.SEVERE, null, ex);
                        } catch (UnavailableException ex) {
                            Logger.getLogger(CMessageMapper.class.getName()).log(Level.SEVERE, null, ex);
                        } catch (TimedOutException ex) {
                            Logger.getLogger(CMessageMapper.class.getName()).log(Level.SEVERE, null, ex);
                        }
                        if (UpdatedFlags.flagsChanged(originalFlags, current)) {
                            // increase the mod-seq as we changed the flags
                            //put.add(MESSAGES_META_CF, MESSAGE_MODSEQ, Bytes.toBytes(modSeq));
                            // update put not to include the allready existing flags
                           byte[] b=CUtils.messageRowKey(member.getMailboxId(),member.getUid());
                            try {
                                d= client.get(ByteBuffer.wrap(b), c, ConsistencyLevel.ONE);
                            } catch (InvalidRequestException ex) {
                                Logger.getLogger(CMessageMapper.class.getName()).log(Level.SEVERE, null, ex);
                            } catch (NotFoundException ex) {
                                Logger.getLogger(CMessageMapper.class.getName()).log(Level.SEVERE, null, ex);
                            } catch (UnavailableException ex) {
                                Logger.getLogger(CMessageMapper.class.getName()).log(Level.SEVERE, null, ex);
                            } catch (TimedOutException ex) {
                                Logger.getLogger(CMessageMapper.class.getName()).log(Level.SEVERE, null, ex);
                            } catch (TException ex) {
                                Logger.getLogger(CMessageMapper.class.getName()).log(Level.SEVERE, null, ex);
                            }
                           d.getColumn().setValue(ByteBufferUtil.bytes(modSeq));
                        }
                        UpdatedFlags uFlags = new UpdatedFlags(member.getUid(), member.getModSeq(), originalFlags, current);
                        updatedFlags.add(uFlags);
                    }

                } catch (IOException e) {
                    throw new MailboxException("Error setting flags for messages in " + mailbox, e);
                } 
            }
        });
        return updatedFlags.iterator();
    }

/*
     * (non-Javadoc)
     * @see org.apache.james.mailbox.store.mail.MessageMapper#copy(org.apache.james.mailbox.store.mail.model.Mailbox, org.apache.james.mailbox.store.mail.model.Message)
     */
    @Override
    public MessageMetaData copy(Mailbox<UUID> mailbox, Message<UUID> original) throws MailboxException {
        long uid = uidProvider.nextUid(mailboxSession, mailbox);
        long modSeq = -1;
        if (modSeqProvider != null) {
            modSeq = modSeqProvider.nextModSeq(mailboxSession, mailbox);
        }
        //TODO: check if creating a HBase message is the right thing to do
        CMessage message = new CMessage(mailbox.getMailboxId(), uid, modSeq, original);
        return save(mailbox, message);
    }

    /*
     * (non-Javadoc)
     * @see org.apache.james.mailbox.store.mail.MessageMapper#getLastUid(org.apache.james.mailbox.store.mail.model.Mailbox)
     */
    @Override
    public long getLastUid(Mailbox<UUID> mailbox) throws MailboxException {
        return uidProvider.lastUid(mailboxSession, mailbox);
    }

    /*
     * (non-Javadoc)
     * @see org.apache.james.mailbox.store.mail.MessageMapper#getHighestModSeq(org.apache.james.mailbox.store.mail.model.Mailbox)
     */
    @Override
    public long getHighestModSeq(Mailbox<UUID> mailbox) throws MailboxException {
        return modSeqProvider.highestModSeq(mailboxSession, mailbox);
    }

    /**
     * Save the {@link Message} for the given {@link Mailbox} and return the {@link MessageMetaData} 
     * 
     * @param mailbox
     * @param message
     * @return metaData
     * @throws MailboxException
     */
    protected MessageMetaData save(Mailbox<UUID> mailbox, Message<UUID> message) throws MailboxException {
        try {
           ColumnPath c=new ColumnPath();
           ColumnOrSuperColumn d = new ColumnOrSuperColumn();
           byte[] a = null;
           client = CUtils.metadataToPut(message);
          //i think we should put client as an argument than keyspace below
                        ChunkOutputStream out = new ChunkOutputStream(CNames.keyspace,ByteBufferUtil.string(CNames.MESSAGE_DATA_BODY), CNames.MESSAGES_DATA,CUtils.messageRowKey(message).toString(),CNames.MAX_COLUMN_SIZE);
            int b;
            BufferedInputStream in = new BufferedInputStream(message.getBodyContent());
            while ((b = in.read()) != -1) {
                out.write(b);
            }
            out.close();
            
            out = new ChunkOutputStream(CNames.keyspace,ByteBufferUtil.string(CNames.MESSAGE_DATA_HEADERS), CNames.MESSAGES_DATA_HEADER,CUtils.messageRowKey(message).toString(),CNames.MAX_COLUMN_SIZE);
            in = new BufferedInputStream(message.getHeaderContent());
            while ((b = in.read()) != -1) {
                out.write(b);
            }
            out.close();
           
            // increase the message count for the current mailbox
           c.setColumn_family(ByteBufferUtil.string(CNames.MAILBOX_CF));
           c.setColumn(CNames.MAILBOX_MESSAGE_COUNT);
           d= client.get(ByteBuffer.wrap(CUtils.mailboxRowKey(mailbox.getMailboxId())), c, ConsistencyLevel.ONE);
           
           a = d.getColumn().getValue();
           d.getColumn().setValue(ByteBufferUtil.bytes(ByteBufferUtil.toLong(ByteBufferUtil.bytes(a.toString())) + 1));
           
        } catch (NotFoundException ex) {
            Logger.getLogger(CMessageMapper.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TTransportException ex) {
            Logger.getLogger(CMessageMapper.class.getName()).log(Level.SEVERE, null, ex);
        } catch (InvalidRequestException ex) {
            Logger.getLogger(CMessageMapper.class.getName()).log(Level.SEVERE, null, ex);
        } catch (CharacterCodingException ex) {
            Logger.getLogger(CMessageMapper.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TException ex) {
            Logger.getLogger(CMessageMapper.class.getName()).log(Level.SEVERE, null, ex);
        } catch (UnavailableException ex) {
            Logger.getLogger(CMessageMapper.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TimedOutException ex) {
            Logger.getLogger(CMessageMapper.class.getName()).log(Level.SEVERE, null, ex);
        } catch (IOException ex) {
            throw new MailboxException("Error setting flags for messages in " + mailbox, ex);
        } 
        
            return new SimpleMessageMetaData(message);
    }

    private void deleteDeletedMessagesInMailboxWithUID(Mailbox<UUID> mailbox, long uid) throws IOException {
        try {
            //TODO: do I have to check if the message is flagged for delete here?
               ColumnPath c=new ColumnPath();
               ColumnOrSuperColumn d = new ColumnOrSuperColumn();
               
               c.setColumn_family(ByteBufferUtil.string(CNames.MESSAGES_META_CF));
               
               c.setColumn(CNames.MESSAGE_BODY_OCTETS);
               d= client.get(ByteBuffer.wrap(CUtils.messageRowKey(mailbox.getMailboxId(), uid)), c, ConsistencyLevel.ONE);
               client.remove(ByteBuffer.wrap(CUtils.messageRowKey(mailbox.getMailboxId(), uid)), c, d.getColumn().getTimestamp(), ConsistencyLevel.ONE);
             
              c.setColumn(CNames.MESSAGE_CONTENT_OCTETS);
               d= client.get(ByteBuffer.wrap(CUtils.messageRowKey(mailbox.getMailboxId(), uid)), c, ConsistencyLevel.ONE);
               client.remove(ByteBuffer.wrap(CUtils.messageRowKey(mailbox.getMailboxId(), uid)), c, d.getColumn().getTimestamp(), ConsistencyLevel.ONE);
               
               c.setColumn(CNames.MESSAGE_INTERNALDATE);
               d= client.get(ByteBuffer.wrap(CUtils.messageRowKey(mailbox.getMailboxId(), uid)), c, ConsistencyLevel.ONE);
               client.remove(ByteBuffer.wrap(CUtils.messageRowKey(mailbox.getMailboxId(), uid)), c, d.getColumn().getTimestamp(), ConsistencyLevel.ONE);
               
               c.setColumn(CNames.MESSAGE_MEDIA_TYPE);
               d= client.get(ByteBuffer.wrap(CUtils.messageRowKey(mailbox.getMailboxId(), uid)), c, ConsistencyLevel.ONE);
               client.remove(ByteBuffer.wrap(CUtils.messageRowKey(mailbox.getMailboxId(), uid)), c, d.getColumn().getTimestamp(), ConsistencyLevel.ONE);
               
               c.setColumn(CNames.MESSAGE_MODSEQ);
               d= client.get(ByteBuffer.wrap(CUtils.messageRowKey(mailbox.getMailboxId(), uid)), c, ConsistencyLevel.ONE);
               client.remove(ByteBuffer.wrap(CUtils.messageRowKey(mailbox.getMailboxId(), uid)), c, d.getColumn().getTimestamp(), ConsistencyLevel.ONE);

               c.setColumn(CNames.MESSAGE_SUB_TYPE);
               d= client.get(ByteBuffer.wrap(CUtils.messageRowKey(mailbox.getMailboxId(), uid)), c, ConsistencyLevel.ONE);
               client.remove(ByteBuffer.wrap(CUtils.messageRowKey(mailbox.getMailboxId(), uid)), c, d.getColumn().getTimestamp(), ConsistencyLevel.ONE);
               
               c.setColumn(CNames.MESSAGE_TEXT_LINE_COUNT);
               d= client.get(ByteBuffer.wrap(CUtils.messageRowKey(mailbox.getMailboxId(), uid)), c, ConsistencyLevel.ONE);
               client.remove(ByteBuffer.wrap(CUtils.messageRowKey(mailbox.getMailboxId(), uid)), c, d.getColumn().getTimestamp(), ConsistencyLevel.ONE);
               
               c.setColumn(CNames.FLAGS_ANSWERED);
               d= client.get(ByteBuffer.wrap(CUtils.messageRowKey(mailbox.getMailboxId(), uid)), c, ConsistencyLevel.ONE);
               client.remove(ByteBuffer.wrap(CUtils.messageRowKey(mailbox.getMailboxId(), uid)), c, d.getColumn().getTimestamp(), ConsistencyLevel.ONE);
               
               c.setColumn(CNames.FLAGS_DELETED);
               d= client.get(ByteBuffer.wrap(CUtils.messageRowKey(mailbox.getMailboxId(), uid)), c, ConsistencyLevel.ONE);
               client.remove(ByteBuffer.wrap(CUtils.messageRowKey(mailbox.getMailboxId(), uid)), c, d.getColumn().getTimestamp(), ConsistencyLevel.ONE);
               
               c.setColumn(CNames.FLAGS_DRAFT);
               d= client.get(ByteBuffer.wrap(CUtils.messageRowKey(mailbox.getMailboxId(), uid)), c, ConsistencyLevel.ONE);
               client.remove(ByteBuffer.wrap(CUtils.messageRowKey(mailbox.getMailboxId(), uid)), c, d.getColumn().getTimestamp(), ConsistencyLevel.ONE);
               
               c.setColumn(CNames.FLAGS_FLAGGED);
               d= client.get(ByteBuffer.wrap(CUtils.messageRowKey(mailbox.getMailboxId(), uid)), c, ConsistencyLevel.ONE);
               client.remove(ByteBuffer.wrap(CUtils.messageRowKey(mailbox.getMailboxId(), uid)), c, d.getColumn().getTimestamp(), ConsistencyLevel.ONE);
               
               c.setColumn(CNames.FLAGS_RECENT);
               d= client.get(ByteBuffer.wrap(CUtils.messageRowKey(mailbox.getMailboxId(), uid)), c, ConsistencyLevel.ONE);
               client.remove(ByteBuffer.wrap(CUtils.messageRowKey(mailbox.getMailboxId(), uid)), c, d.getColumn().getTimestamp(), ConsistencyLevel.ONE);
   
               c.setColumn(CNames.FLAGS_SEEN);
               d= client.get(ByteBuffer.wrap(CUtils.messageRowKey(mailbox.getMailboxId(), uid)), c, ConsistencyLevel.ONE);
               client.remove(ByteBuffer.wrap(CUtils.messageRowKey(mailbox.getMailboxId(), uid)), c, d.getColumn().getTimestamp(), ConsistencyLevel.ONE);
               
               c.setColumn(CNames.FLAGS_USER);
               d= client.get(ByteBuffer.wrap(CUtils.messageRowKey(mailbox.getMailboxId(), uid)), c, ConsistencyLevel.ONE);
               client.remove(ByteBuffer.wrap(CUtils.messageRowKey(mailbox.getMailboxId(), uid)), c, d.getColumn().getTimestamp(), ConsistencyLevel.ONE);
        
               
               c.setColumn_family(ByteBufferUtil.string(CNames.MESSAGE_DATA_BODY));

               c.setColumn(ByteBufferUtil.bytes(CNames.MESSAGES_DATA));
               d= client.get(ByteBuffer.wrap(CUtils.messageRowKey(mailbox.getMailboxId(), uid)), c, ConsistencyLevel.ONE);
               client.remove(ByteBuffer.wrap(CUtils.messageRowKey(mailbox.getMailboxId(), uid)), c, d.getColumn().getTimestamp(), ConsistencyLevel.ONE);
               
               c.setColumn_family(ByteBufferUtil.string(CNames.MESSAGE_DATA_HEADERS));
               
               c.setColumn(ByteBufferUtil.bytes(CNames.MESSAGES_DATA_HEADER));
               d= client.get(ByteBuffer.wrap(CUtils.messageRowKey(mailbox.getMailboxId(), uid)), c, ConsistencyLevel.ONE);
               client.remove(ByteBuffer.wrap(CUtils.messageRowKey(mailbox.getMailboxId(), uid)), c, d.getColumn().getTimestamp(), ConsistencyLevel.ONE);
               
               c.setColumn_family(ByteBufferUtil.string(CNames.MAILBOX_CF));
               c.setColumn(CNames.MAILBOX_MESSAGE_COUNT);
               d= client.get(ByteBuffer.wrap(CUtils.mailboxRowKey(mailbox.getMailboxId())), c, ConsistencyLevel.ONE);
               byte[] a = d.getColumn().getValue();
               d.getColumn().setValue(ByteBufferUtil.bytes(ByteBufferUtil.toLong(ByteBufferUtil.bytes(a.toString())) - 1));
               
               c.setColumn(CNames.MAILBOX_HIGHEST_MODSEQ);
               d= client.get(ByteBuffer.wrap(CUtils.mailboxRowKey(mailbox.getMailboxId())), c, ConsistencyLevel.ONE);
               a = d.getColumn().getValue();
               d.getColumn().setValue(ByteBufferUtil.bytes(ByteBufferUtil.toLong(ByteBufferUtil.bytes(a.toString())) + 1));
               
        } catch (InvalidRequestException ex) {
            Logger.getLogger(CMessageMapper.class.getName()).log(Level.SEVERE, null, ex);
        } catch (NotFoundException ex) {
            Logger.getLogger(CMessageMapper.class.getName()).log(Level.SEVERE, null, ex);
        } catch (UnavailableException ex) {
            Logger.getLogger(CMessageMapper.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TimedOutException ex) {
            Logger.getLogger(CMessageMapper.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TException ex) {
            Logger.getLogger(CMessageMapper.class.getName()).log(Level.SEVERE, null, ex);
        }
           
           
        
    }

    private void deleteDeletedMessagesInMailboxBetweenUIDs(Mailbox<UUID> mailbox, long fromUid, long toUid) throws IOException {
       
    }

    private void deleteDeletedMessagesInMailboxAfterUID(Mailbox<UUID> mailbox, long fromUid) throws IOException {
        
    }

    private void deleteDeletedMessagesInMailbox(Mailbox<UUID> mailbox) throws IOException {
        try {
            ColumnPath c=new ColumnPath();
            count=0;
             ColumnOrSuperColumn d = new ColumnOrSuperColumn();
             c.setColumn_family(ByteBufferUtil.string(CNames.MESSAGES_META_CF));
             c.setColumn(CNames.FLAGS_DELETED); 
             
             ColumnParent messages= new ColumnParent(ByteBufferUtil.string(CNames.MESSAGES_META_CF)); 
             SlicePredicate predicate = new SlicePredicate();
             predicate.setSlice_range(new SliceRange(ByteBuffer.wrap(new byte[0]), ByteBuffer.wrap(new byte[0]), false, 100));
             KeyRange keyRange = new KeyRange(100);
             keyRange.setStart_key(CUtils.messageRowKey(mailbox.getMailboxId(), 0));
              keyRange.setEnd_key(CUtils.messageRowKey(mailbox.getMailboxId(),Long.MAX_VALUE));
              List<KeySlice> keySlices = client.get_range_slices(messages, predicate, keyRange, ConsistencyLevel.ONE);
              
              for (KeySlice ks : keySlices) {
                  d = client.get(ByteBuffer.wrap(ks.getKey()), c, ConsistencyLevel.ONE);
                  byte[] b = d.getColumn().getValue();
                  if(ByteBufferUtil.compare(b, CNames.MARKER_PRESENT)==0){
                      c.setColumn_family(ByteBufferUtil.string(CNames.MESSAGES_META_CF));
             
             c.setColumn(CNames.MESSAGE_BODY_OCTETS);
             d= client.get(ByteBuffer.wrap(ks.getKey()), c, ConsistencyLevel.ONE);
             client.remove(ByteBuffer.wrap(ks.getKey()), c, d.getColumn().getTimestamp(), ConsistencyLevel.ONE);
           
            c.setColumn(CNames.MESSAGE_CONTENT_OCTETS);
             d= client.get(ByteBuffer.wrap(ks.getKey()), c, ConsistencyLevel.ONE);
             client.remove(ByteBuffer.wrap(ks.getKey()), c, d.getColumn().getTimestamp(), ConsistencyLevel.ONE);
             
             c.setColumn(CNames.MESSAGE_INTERNALDATE);
             d= client.get(ByteBuffer.wrap(ks.getKey()), c, ConsistencyLevel.ONE);
             client.remove(ByteBuffer.wrap(ks.getKey()), c, d.getColumn().getTimestamp(), ConsistencyLevel.ONE);
             
             c.setColumn(CNames.MESSAGE_MEDIA_TYPE);
             d= client.get(ByteBuffer.wrap(ks.getKey()), c, ConsistencyLevel.ONE);
             client.remove(ByteBuffer.wrap(ks.getKey()), c, d.getColumn().getTimestamp(), ConsistencyLevel.ONE);
             
             c.setColumn(CNames.MESSAGE_MODSEQ);
             d= client.get(ByteBuffer.wrap(ks.getKey()), c, ConsistencyLevel.ONE);
             client.remove(ByteBuffer.wrap(ks.getKey()), c, d.getColumn().getTimestamp(), ConsistencyLevel.ONE);

             c.setColumn(CNames.MESSAGE_SUB_TYPE);
             d= client.get(ByteBuffer.wrap(ks.getKey()), c, ConsistencyLevel.ONE);
             client.remove(ByteBuffer.wrap(ks.getKey()), c, d.getColumn().getTimestamp(), ConsistencyLevel.ONE);
             
             c.setColumn(CNames.MESSAGE_TEXT_LINE_COUNT);
             d= client.get(ByteBuffer.wrap(ks.getKey()), c, ConsistencyLevel.ONE);
             client.remove(ByteBuffer.wrap(ks.getKey()), c, d.getColumn().getTimestamp(), ConsistencyLevel.ONE);
             
             c.setColumn(CNames.FLAGS_ANSWERED);
             d= client.get(ByteBuffer.wrap(ks.getKey()), c, ConsistencyLevel.ONE);
             client.remove(ByteBuffer.wrap(ks.getKey()), c, d.getColumn().getTimestamp(), ConsistencyLevel.ONE);
             
             c.setColumn(CNames.FLAGS_DELETED);
             d= client.get(ByteBuffer.wrap(ks.getKey()), c, ConsistencyLevel.ONE);
             client.remove(ByteBuffer.wrap(ks.getKey()), c, d.getColumn().getTimestamp(), ConsistencyLevel.ONE);
             
             c.setColumn(CNames.FLAGS_DRAFT);
             d= client.get(ByteBuffer.wrap(ks.getKey()), c, ConsistencyLevel.ONE);
             client.remove(ByteBuffer.wrap(ks.getKey()), c, d.getColumn().getTimestamp(), ConsistencyLevel.ONE);
             
             c.setColumn(CNames.FLAGS_FLAGGED);
             d= client.get(ByteBuffer.wrap(ks.getKey()), c, ConsistencyLevel.ONE);
             client.remove(ByteBuffer.wrap(ks.getKey()), c, d.getColumn().getTimestamp(), ConsistencyLevel.ONE);
             
             c.setColumn(CNames.FLAGS_RECENT);
             d= client.get(ByteBuffer.wrap(ks.getKey()), c, ConsistencyLevel.ONE);
             client.remove(ByteBuffer.wrap(ks.getKey()), c, d.getColumn().getTimestamp(), ConsistencyLevel.ONE);
             
             c.setColumn(CNames.FLAGS_SEEN);
             d= client.get(ByteBuffer.wrap(ks.getKey()), c, ConsistencyLevel.ONE);
             client.remove(ByteBuffer.wrap(ks.getKey()), c, d.getColumn().getTimestamp(), ConsistencyLevel.ONE);
             
             c.setColumn(CNames.FLAGS_USER);
             d= client.get(ByteBuffer.wrap(ks.getKey()), c, ConsistencyLevel.ONE);
             client.remove(ByteBuffer.wrap(ks.getKey()), c, d.getColumn().getTimestamp(), ConsistencyLevel.ONE);
      
             
             c.setColumn_family(ByteBufferUtil.string(CNames.MESSAGE_DATA_BODY));

             c.setColumn(ByteBufferUtil.bytes(CNames.MESSAGES_DATA));
             d= client.get(ByteBuffer.wrap(ks.getKey()), c, ConsistencyLevel.ONE);
             client.remove(ByteBuffer.wrap(ks.getKey()), c, d.getColumn().getTimestamp(), ConsistencyLevel.ONE);
             
             c.setColumn_family(ByteBufferUtil.string(CNames.MESSAGE_DATA_HEADERS));
             
             c.setColumn(ByteBufferUtil.bytes(CNames.MESSAGES_DATA_HEADER));
             d= client.get(ByteBuffer.wrap(ks.getKey()), c, ConsistencyLevel.ONE);
             client.remove(ByteBuffer.wrap(ks.getKey()), c, d.getColumn().getTimestamp(), ConsistencyLevel.ONE);
             
               count++;       
              }
          }
             c.setColumn_family(ByteBufferUtil.string(CNames.MAILBOX_CF));
             c.setColumn(CNames.MAILBOX_MESSAGE_COUNT);
             d= client.get(ByteBuffer.wrap(CUtils.mailboxRowKey(mailbox.getMailboxId())), c, ConsistencyLevel.ONE);
             byte[] a = d.getColumn().getValue();
             d.getColumn().setValue(ByteBufferUtil.bytes(ByteBufferUtil.toLong(ByteBufferUtil.bytes(a.toString())) - count));
             
             c.setColumn(CNames.MAILBOX_HIGHEST_MODSEQ);
             d= client.get(ByteBuffer.wrap(CUtils.mailboxRowKey(mailbox.getMailboxId())), c, ConsistencyLevel.ONE);
             a = d.getColumn().getValue();
             d.getColumn().setValue(ByteBufferUtil.bytes(ByteBufferUtil.toLong(ByteBufferUtil.bytes(a.toString())) + 1));
        } catch (NotFoundException ex) {
            Logger.getLogger(CMessageMapper.class.getName()).log(Level.SEVERE, null, ex);
        } catch (InvalidRequestException ex) {
            Logger.getLogger(CMessageMapper.class.getName()).log(Level.SEVERE, null, ex);
        } catch (UnavailableException ex) {
            Logger.getLogger(CMessageMapper.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TimedOutException ex) {
            Logger.getLogger(CMessageMapper.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TException ex) {
            Logger.getLogger(CMessageMapper.class.getName()).log(Level.SEVERE, null, ex);
        }
    }
    private Map<Long, MessageMetaData> createMetaData(List<Message<UUID>> uids) {
        final Map<Long, MessageMetaData> data2 = new HashMap<Long, MessageMetaData>();
        for (int i = 0; i < uids.size(); i++) {
            Message<UUID> m = uids.get(i);
            data2.put(m.getUid(), new SimpleMessageMetaData(m));
        }
        return data2;
    }

}
